package com.atguigu.flink.state;

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by 黄凯 on 2023/6/25 0025 16:21
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 通用增量(目前是实验性功能)
 * *
 * * 1. RocksDB开启增量检查点
 */
public class Flink01_ChangeLog {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //代码中设置
        //配置RocksDB状态后端，开启增量检查点
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

        //配置文件中设置
        // state.backend.incremental: true

        //代码中设置
        //可以通过开启changlog，让hashmap和rocksdb支持增量检查点。
        env.enableChangelogStateBackend(true);

        //配置文件中设置
        //state.backend.changelog.enabled: true
        //state.backend.changelog.storage: filesystem
        //# 存储 changelog 数据
        //dstl.dfs.base-path: hdfs://hadoop102:8020/changelog
        //execution.checkpointing.max-concurrent-checkpoints: 1
        //execution.savepoint-restore-mode: CLAIM

        env.execute();

    }

}
